package com.djk.daily.kafka.producer;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

public class ProducerSimple {
    public final static String TOPIC_NAME = "djk_topic";

    public static void main(String[] args) throws Exception{

        //异步发送
//        producerSend();
        //同步发送
//        producerAsynSend();
        //Producer异步发送回调演示
//        producerSendWithCallBack();
        producerSendWithCallBackAndPartition();
    }

    /**
     * Producer异步发送演示
     */
    public static void producerSend(){
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"101.35.134.133:9092");
        properties.put(ProducerConfig.ACKS_CONFIG,"all");
        properties.put(ProducerConfig.RETRIES_CONFIG,"0");
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG,"16384");
        properties.put(ProducerConfig.LINGER_MS_CONFIG,"1");
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,"33554432");

        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");

        //Producer的主对象
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        //消息对象 - ProducerRecord
        for (int i = 10; i < 20; i++) {
            ProducerRecord<String,String> record =
                    new ProducerRecord<>(TOPIC_NAME,"key-" + i,"value-" + i);
            producer.send(record);
        }

        producer.close();
    }

    /**
     * Producer异步发送回调演示
     */
    public static void producerSendWithCallBack(){
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"101.35.134.133:9092");
        properties.put(ProducerConfig.ACKS_CONFIG,"all");
        properties.put(ProducerConfig.RETRIES_CONFIG,"0");
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG,"16384");
        properties.put(ProducerConfig.LINGER_MS_CONFIG,"1");
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,"33554432");

        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");

        //Producer的主对象
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        //消息对象 - ProducerRecord
        for (int i = 10; i < 20; i++) {
            ProducerRecord<String,String> record =
                    new ProducerRecord<>(TOPIC_NAME,"key-" + i,"value-" + i);
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    System.out.println("partition : " + recordMetadata.partition() + ",offset : " + recordMetadata.offset());
                }
            });
        }

        producer.close();
    }


    /**
     * Producer同步发送演示（异步阻塞）
     */
    public static void producerAsynSend() throws Exception {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"101.35.134.133:9092");
        properties.put(ProducerConfig.ACKS_CONFIG,"all");
        properties.put(ProducerConfig.RETRIES_CONFIG,"0");
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG,"16384");
        properties.put(ProducerConfig.LINGER_MS_CONFIG,"1");
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,"33554432");

        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");

        //Producer的主对象
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        //消息对象 - ProducerRecord
        for (int i = 30; i < 40; i++) {
            ProducerRecord<String,String> record =
                    new ProducerRecord<>(TOPIC_NAME,"key-" + i,"value-" + i);
            Future<RecordMetadata> send = producer.send(record);
            //获取发送结果的时候会阻塞下
            RecordMetadata recordMetadata = send.get();
            System.out.println("partition : " + recordMetadata.partition() + ",offset : " + recordMetadata.offset());
        }

        //所有的通道打开都需要全部关闭
        producer.close();
    }

    /**
     * Producer异步发送回调演示,同时自定义partition
     */
    public static void producerSendWithCallBackAndPartition(){
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"101.35.134.133:9092");
        properties.put(ProducerConfig.ACKS_CONFIG,"all");
        properties.put(ProducerConfig.RETRIES_CONFIG,"0");
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG,"16384");
        properties.put(ProducerConfig.LINGER_MS_CONFIG,"1");
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,"33554432");

        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.djk.daily.kafka.producer.SimplePartition");

        //Producer的主对象
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        //消息对象 - ProducerRecord
        for (int i = 20; i < 120; i++) {
            ProducerRecord<String,String> record =
                    new ProducerRecord<>(TOPIC_NAME,"key-" + i,"value-" + i);
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    System.out.println("partition : " + recordMetadata.partition() + ",offset : " + recordMetadata.offset());
                }
            });
        }

        producer.close();
    }
}
